package org.test.service;

import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.state.SingleArcTransition;
import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;

public class JobStateMachine implements EventHandler<JobEvent> {

	private String jobId;
	private EventHandler<JobEvent> enentHandler;
	
	private StateMachine<JobStateInternal, JobEventType, JobEvent> stateMachine;
	
	private StateMachineFactory<JobStateMachine, JobStateInternal, JobEventType, JobEvent> factory
		= new StateMachineFactory<JobStateMachine, JobStateInternal, JobEventType, JobEvent>(JobStateInternal.NEW)
		.addTransition(JobStateInternal.NEW, JobStateInternal.RUNNING, JobEventType.BEGIN,new BeginTransaction())
		.addTransition(JobStateInternal.RUNNING, JobStateInternal.STOPPED, JobEventType.END,new EndTransaction())
		.installTopology();
	
	public JobStateMachine(String jobId,EventHandler<JobEvent> eventHandler){
		this.jobId = jobId;
		this.enentHandler = eventHandler;
		stateMachine = factory.make(this);
	}
	

	public void handle(JobEvent event) {
		JobStateInternal old = getJobStateInternal();
		getStateMachine().doTransition(event.getType(), event);
		System.out.println("Job trandsactioned from :"+old+"to:"+getJobStateInternal());
	}
	
	

	public StateMachine<JobStateInternal, JobEventType, JobEvent> getStateMachine() {
		return stateMachine;
	}

	public JobStateInternal getJobStateInternal(){
		return getStateMachine().getCurrentState();
	}

	private class BeginTransaction implements SingleArcTransition<JobStateMachine, JobEvent>{

		public void transition(JobStateMachine job, JobEvent event) {
			System.out.println("receive event : "+event.getType());
			//job.enentHandler.handle(new JobEvent(event.getJobId(), JobEventType.BEGIN));
		}
		
	}
	
	private class EndTransaction implements SingleArcTransition<JobStateMachine, JobEvent>{

		public void transition(JobStateMachine job, JobEvent event) {
			System.out.println("receive event : "+event.getType());
			//job.enentHandler.handle(new JobEvent(event.getJobId(), JobEventType.END));
		}
		
	}
}
